package org.ider.qiyu.live.id.generate.interfaces.service.Impl;

import jakarta.annotation.Resource;
import org.ider.qiyu.live.id.generate.interfaces.dao.mapper.IdGenerateMapper;
import org.ider.qiyu.live.id.generate.interfaces.dao.po.IdGeneratePO;
import org.ider.qiyu.live.id.generate.interfaces.service.IdGenerateService;
import org.ider.qiyu.live.id.generate.interfaces.service.bo.LocalSeqIdBO;
import org.ider.qiyu.live.id.generate.interfaces.service.bo.LocalUnSeqIdBO;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;

/**
 * @Description: //todo
 * @Author: 张广焱
 * @Date: 2024-04-04 8:28
 */
@Service
public class IdGenerateServiceImpl implements IdGenerateService, InitializingBean {
    private static final Logger log = LoggerFactory.getLogger(IdGenerateServiceImpl.class);
    @Resource
    private IdGenerateMapper idGenerateMapper;
    private static Map<Integer, Semaphore> semaphoreMap = new ConcurrentHashMap<>();

    private static ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(8, 16, 3, TimeUnit.SECONDS, new LinkedBlockingQueue<>(1000),
            new ThreadFactory() {
                @Override
                public Thread newThread(Runnable r) {
                    Thread thread = new Thread(r);
                    thread.setName("id-generate-thread-" + ThreadLocalRandom.current().nextInt(1000));
                    return thread;
                }
            });

    /**
     * @Description: 有序idMap
     * @Author: 张广焱
     * @Date: 2024-04-04  15:48:49
     */
    private static Map<Integer, LocalSeqIdBO> locakSeqIdBoMap = new ConcurrentHashMap<>();
    /**
     * @Description: 无序idMap
     * @Param:
     * @Author: 张广焱
     * @Date: 2024-04-04  15:49:06
     */
    private static Map<Integer, LocalUnSeqIdBO> locakUnSeqIdBoMap = new ConcurrentHashMap<>();

    //阈值常量
    private static final float UPDATE_RATE = 0.75f;
    private static final int SEQ_ID = 1;

    @Override
    public Long getSeqId(Integer id) {
        if (id == null) {
            log.error("[getseqId] id is error ,id is {}", id);
            return null;
        }
        LocalSeqIdBO locakSeqIdBo = locakSeqIdBoMap.get(id);
        if (locakSeqIdBo == null) {
            log.error("[getseqId] locakSeqIdBo is null ,id is {}", id);
            return null;
        }
        //设置阈值比对
        this.refreshLoaclSeqId(locakSeqIdBo);
        //原子类 增1 保证线程安全
        //dubbo 服务只要是靠异步线程池
        long returnId = locakSeqIdBo.getCurrentValue().incrementAndGet();
        if ( returnId> locakSeqIdBo.getNextThreshold()) {
            //同步去刷新id 没有什么必要 直接失败
            log.error("[getseqId] id is error ,id is {}", id);
            return null;
        }
        return returnId;
    }

    /**
     * @Description:更新mysql里面的分布式id配置信息,占用响应的id段
     * @Param: @param localSeqIdBO
     * @Author: 张广焱
     * @Date: 2024-04-04  10:41:22
     */
    private void refreshLoaclSeqId(LocalSeqIdBO localSeqIdBO) {
        //这个id段的阈值 用了多少id
        long step = localSeqIdBO.getNextThreshold() - localSeqIdBO.getCurrentStart();
        if (localSeqIdBO.getCurrentValue().get() - localSeqIdBO.getCurrentStart() > step * UPDATE_RATE) {
            Semaphore semaphore = semaphoreMap.get(localSeqIdBO.getId());
            if (semaphore == null) {
                log.error("[refreshLoaclSeqId] semaphoreMap is null", localSeqIdBO.getId());
                return;
            }
            boolean acquireStatus = semaphore.tryAcquire();
            if (acquireStatus) {
                log.info("尝试开始进行同步同步更新");
                //异步更新id
                threadPoolExecutor.execute(new Runnable() {
                    @Override
                    public void run() {
                        try {
                            //重新刷新id
                            IdGeneratePO idGeneratePO = idGenerateMapper.selectById(localSeqIdBO.getId());
                            tryUpdateMySQLRecord(idGeneratePO);
                        } catch (Exception e) {
                            log.error("[refreshLoaclSeqId] updateIdGenerate error", e);
                        }finally {
                            //保证线程安全 释放信号量
                            semaphoreMap.get(localSeqIdBO.getId()).release();
                            log.info("本地有序id同步段完成,id is {}", localSeqIdBO.getId());
                        }
                    }
                });
            }
        }

    }

    /**
     * @Description: 刷新本地无序id
     * @Param: @param localUnSeqIdBO
     * @Author: 张广焱
     * @Date: 2024-04-04  16:47:12
     */
    private void refreshLoaclUnSeqId(LocalUnSeqIdBO localUnSeqIdBO) {
        Long begin = localUnSeqIdBO.getCurrentStart();
        Long end = localUnSeqIdBO.getNextThreshold();
        int remainSize = localUnSeqIdBO.getIdQueue().size();
        //如果剩余的id数量小于阈值的25% 就去刷新id
        if ((end - begin) * 0.25 > remainSize) {
            Semaphore semaphore = semaphoreMap.get(localUnSeqIdBO.getId());
            if (semaphore == null) {
                log.error("[refreshLoaclUnSeqId] semaphoreMap is null", localUnSeqIdBO.getId());
                return;
            }
            boolean acquireStatus = semaphore.tryAcquire();
            if (acquireStatus) {
                threadPoolExecutor.execute(new Runnable() {
                    @Override
                    public void run() {
                        try {
                            IdGeneratePO idGeneratePO = idGenerateMapper.selectById(localUnSeqIdBO.getId());
                            tryUpdateMySQLRecord(idGeneratePO);
                        }catch (Exception e){
                            log.error("[refreshLoaclUnSeqId] updateIdGenerate error", e);
                        }finally {
                            //保证线程安全 释放信号量
                            semaphoreMap.get(localUnSeqIdBO.getId()).release();
                            log.info("本地有序id同步段完成,id is {}", localUnSeqIdBO.getId());
                        }
                    }
                });
            }
        }
    }

    @Override
    public Long getUnSeqId(Integer id) {
        if (id == null) {
            log.error("[getUnSeqId] id is error ,id is {}", id);
            return null;
        }
        LocalUnSeqIdBO localUnSeqIdBO = locakUnSeqIdBoMap.get(id);
        if (localUnSeqIdBO == null) {
            log.error("[getUnSeqId] localUnSeqIdBO is null ,id is {}", id);
            return null;
        }
        //线程池队列 取值
        Long returnId = localUnSeqIdBO.getIdQueue().poll();
        if (returnId == null) {
            //同步去刷新id 没有什么必要 直接失败
            log.error("[getUnSeqId] returnId is error ,id is {}", id);
            return null;
        }
        this.refreshLoaclUnSeqId(localUnSeqIdBO);
        return returnId;
    }


    /**
     * @Description: 启动springboot的时候 初始化；bean初始化的时候  会调用这个方法
     * @Param:
     * @Author: 张广焱
     * @Date: 2024-04-04  08:41:12
     */
    @Override
    public void afterPropertiesSet() throws Exception {
        List<IdGeneratePO> idGeneratePOList = idGenerateMapper.selectAll();
        for (IdGeneratePO idGeneratePO : idGeneratePOList) {
            log.info("服务刚启动,抢占新的id段");
            this.tryUpdateMySQLRecord(idGeneratePO);
            //一次只允许一个线程去通过
            semaphoreMap.put(idGeneratePO.getId(), new Semaphore(1));
        }
    }

    private void tryUpdateMySQLRecord(IdGeneratePO idBuilderPO) throws RuntimeException {

        int updateResult = idGenerateMapper.updateNewVersion(idBuilderPO.getId());
        if (updateResult > 0) {
            this.localIdBOHandler(idBuilderPO);
            return;
        }
        for (int i = 0; i < 3; i++) {
            IdGeneratePO idGeneratePO = idGenerateMapper.selectById(idBuilderPO.getId());
            updateResult = idGenerateMapper.updateNewVersion(idGeneratePO.getId());
            if (updateResult > 0) {
                this.localIdBOHandler(idBuilderPO);
                return;
            }
        }
        throw new RuntimeException("表id段占用失败,竞争过于激烈,id is {}" + idBuilderPO.getId());
    }

    /**
     * @Description: 处理如何将本地id对象存放到map中, 并且进行初始化
     * @Param: @param idGeneratePO
     * @Author: 张广焱
     * @Date: 2024-04-04  16:42:05
     */
    private void localIdBOHandler(IdGeneratePO idGeneratePO) {
        long currentStart = idGeneratePO.getCurrentStart();
        long nextThreshold = idGeneratePO.getNextThreshold();
        long currentNum = currentStart;
        if (idGeneratePO.getIsSeq() == SEQ_ID) {
            LocalSeqIdBO localSeqIdBO = new LocalSeqIdBO();
            AtomicLong atomicLong = new AtomicLong(currentNum);
            localSeqIdBO.setCurrentValue(atomicLong);
            localSeqIdBO.setId(idGeneratePO.getId());
            localSeqIdBO.setCurrentStart(currentStart);
            localSeqIdBO.setNextThreshold(nextThreshold);
            locakSeqIdBoMap.put(idGeneratePO.getId(), localSeqIdBO);
        } else {
            LocalUnSeqIdBO localUnSeqIdBO = new LocalUnSeqIdBO();
            localUnSeqIdBO.setCurrentStart(currentStart);
            localUnSeqIdBO.setNextThreshold(nextThreshold);
            localUnSeqIdBO.setId(idGeneratePO.getId());
            //设置id队列
            long begin = localUnSeqIdBO.getCurrentStart();
            long end = localUnSeqIdBO.getNextThreshold();
            List<Long> idList = new ArrayList<>();
            for (long i = begin; i < end; i++) {
                idList.add(i);
            }
            //洗牌算法 将里面的东西随机打乱
            Collections.shuffle(idList);
            ConcurrentLinkedQueue<Long> idQueue = new ConcurrentLinkedQueue<>();
            idQueue.addAll(idList);
            localUnSeqIdBO.setIdQueue(idQueue);
            locakUnSeqIdBoMap.put(idGeneratePO.getId(), localUnSeqIdBO);
        }
    }
}
